package com.zetatech.bdp.rule;

import com.zetatech.bdp.entity.Field;
import com.zetatech.bdp.entity.HiveWriter;
import com.zetatech.bdp.entity.RdbmsWriter;
import com.zetatech.bdp.entity.Rule;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.dom4j.Attribute;
import org.dom4j.Document;
import org.dom4j.DocumentHelper;
import org.dom4j.Element;
import org.dom4j.Node;

import java.io.File;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author cheng.cao@zetatech.com.cn
 */
@Slf4j
public class RuleParser {

    private static final Map<String, Rule> rulesHolder = new ConcurrentHashMap<>();

    public static Rule getFields(String key) {
        return rulesHolder.get(key);
    }

    public static Set<String> getKeys() {
        return rulesHolder.keySet();
    }

    private static String safeGetAttributeVal(Node node, String attr) {
        Node n = node.selectSingleNode("./@" + attr);
        return n == null ? null : ((Attribute) n).getValue();
    }

    private static String safeGetNodeVal(Node node, String xpath) {
        Node n = node.selectSingleNode(xpath);
        return n == null ? null : n.getStringValue();
    }

    private static boolean parseWriter(Node n, Rule r) {
        String writer = safeGetNodeVal(n, "./settings/writer/text()");
        if (writer.equalsIgnoreCase("mysql") || writer.equalsIgnoreCase("oracle")) {
            String url = safeGetNodeVal(n, String.format("./settings/%s/url/text()", writer.trim().toLowerCase()));
            String driver = safeGetNodeVal(n, String.format("./settings/%s/driver/text()", writer.trim().toLowerCase()));
            String username = safeGetNodeVal(n, String.format("./settings/%s/username/text()", writer.trim().toLowerCase()));
            String password = safeGetNodeVal(n, String.format("./settings/%s/password/text()", writer.trim().toLowerCase()));
            String table = safeGetNodeVal(n, String.format("./settings/%s/table/text()", writer.trim().toLowerCase()));
            boolean validateResult = validateParam(url, "url") &&
                    validateParam(driver, "driver") &&
                    validateParam(username, "username") &&
                    validateParam(password, "password") &&
                    validateParam(table, "table");
            RdbmsWriter w = new RdbmsWriter();
            w.setUrl(url);
            w.setDriver(driver);
            w.setUsername(username);
            w.setPassword(password);
            w.setTable(table);
            r.setWriter(w);
            return validateResult;
        } else if (writer.equalsIgnoreCase("hive")) {
            String metastoreUris = safeGetNodeVal(n, "./settings/hive/metastoreUris");
            String defaultFs = safeGetNodeVal(n, "./settings/hive/defaultFs/text()");
            String database = safeGetNodeVal(n, "./settings/hive/database/text()");
            String table = safeGetNodeVal(n, "./settings/hive/table/text()");

            Element confElement = (Element) n.selectSingleNode("./settings/hive/conf");
            Map<String, String> conf = new HashMap<>();
            if (confElement != null) {
                confElement.elements().stream().forEach(e -> conf.put(e.getName(), e.getTextTrim()));
            }
            boolean validateResult = validateParam(metastoreUris, "metastoreUris") &&
                    validateParam(defaultFs, "defaultFs") &&
                    validateParam(database, "database") &&
                    validateParam(table, "table");
            HiveWriter w = new HiveWriter();
            w.setHiveMetastoreUris(metastoreUris);
            w.setDefaultFS(defaultFs);
            w.setDatabase(database);
            w.setTable(table);
            w.setConf(conf);
            r.setWriter(w);
            return validateResult;

        } else if (writer.equalsIgnoreCase("hbase")) {
            return false;
        }
        return false;
    }

    private static boolean validateParam(String username, String logName) {
        if (username == null || username.trim().isEmpty()) {
            log.error("writer下的{}不能未空", logName);
            return false;
        }
        return true;
    }

    public static void load(String configFile) throws Exception {
        String content = FileUtils.readFileToString(new File(configFile), StandardCharsets.UTF_8);
        Document document = DocumentHelper.parseText(content);
        Element root = document.getRootElement();
        Node node = root.selectSingleNode("/rules");
        String namespace = safeGetAttributeVal(node, "namespace");
        String globalEnable = safeGetAttributeVal(node, "enable");
        // 如果不启用的话就直接跳过了
        if ("false".equalsIgnoreCase(globalEnable)) {
            return;
        }
        List<Node> parserList = node.selectNodes("./rule");
        for (Node n : parserList) {
            String itemEnable = safeGetAttributeVal(n, "enable");
            if ("false".equalsIgnoreCase(itemEnable)) {
                continue;
            }
            String id = safeGetAttributeVal(n, "id");
            String parseClazz = safeGetAttributeVal(n, "class");
            String parseType = safeGetAttributeVal(n, "type");
            String writer = safeGetNodeVal(n, "./settings/writer/text()");
            if (writer == null || writer.trim().isEmpty()) {
                log.error("文件:{}中节点[rule/settings/writer],缺少值,忽略掉该条规则", configFile);
                continue;
            }
            writer = writer.trim().toLowerCase();
            if (
                    !(writer.equalsIgnoreCase("mysql") ||
                            writer.equalsIgnoreCase("oracle") ||
                            writer.equalsIgnoreCase("hive") ||
                            writer.equalsIgnoreCase("hbase"))
            ) {
                log.error("文件:{}中节点[rule/settings/writer],writer={}值错误,忽略掉该条规则", configFile, writer);
                continue;
            }
            if ((parseClazz == null || parseClazz.trim().isEmpty()) && (parseType == null || parseType.trim().isEmpty())) {
                log.error("文件:{}中节点[rule],缺少属性type和class(两者必有其一),忽略掉该条规则", configFile);
                continue;
            }
            Rule rule = new Rule();
            if (parseClazz != null && !parseClazz.trim().isEmpty()) {
                rule.setClazz(parseClazz);
            } else {
                if ("simple".equalsIgnoreCase(parseType)) {
                    rule.setClazz("com.zetatech.bdp.parser.processor.impl.SimpleLineProcessor");
                } else if ("default".equalsIgnoreCase(parseType)) {
                    rule.setClazz("com.zetatech.bdp.parser.processor.impl.DeviceLogLineProcessor");
                } else {
                    log.error("文件:{}中节点[rule],不支持的type值，type={},忽略掉该条规则", configFile, parseType);
                    continue;
                }
            }
            if (!parseWriter(n, rule)) {
                continue;
            }
            List<Node> mappingFields = n.selectNodes("./mapping/item");
            List<Field> fields = new ArrayList<>();
            for (Node f : mappingFields) {
                Field field = new Field();
                String property = safeGetAttributeVal(f, "property");
                String column = safeGetAttributeVal(f, "column");
                String type = safeGetAttributeVal(f, "type");
                String prefix = safeGetAttributeVal(f, "prefix");
                String scriptEngine = safeGetAttributeVal(f, "scriptEngine");
                String script = f.getStringValue();
                field.setValue(safeGetAttributeVal(f, "value"));
                field.setScript(script);
                field.setScriptEngine(scriptEngine);
                if (property == null) {
                    throw new RuntimeException("配置文件错误:property不能未空");
                }
                if (prefix != null) {
                    field.setPrefix(prefix);
                }
                field.setProperty(property);
                // 默认column与property相同
                if (column != null) {
                    field.setColumn(column);
                } else {
                    //复杂类型，字段映射默认规则
                    if (field.getPrefix() != null) {
                        String prefixStr = field.getPrefix().replaceAll("-", "_");
                        // 按照文档上给的演例，如果以name开头的复杂类型字段会将name去掉
                        if (prefixStr.toLowerCase().startsWith("name:")) {
                            prefixStr = prefixStr.substring("name:".length());
                        }
                        //替换：为下划线
                        if (prefixStr.endsWith(":") || prefixStr.endsWith(" ") || prefixStr.endsWith(" ")) {
                            prefixStr = prefixStr.substring(0, prefixStr.length() - 1);
                        }

                        prefixStr = prefixStr.replace(":", "_");
                        if (prefixStr.toLowerCase().endsWith(field.getProperty().toLowerCase())) {
                            field.setColumn(prefixStr);
                        } else {
                            if (field.getProperty().endsWith(".")) {
                                field.setColumn(prefixStr + "_" + field.getProperty().substring(0, field.getProperty().length() - 1));
                            } else {
                                field.setColumn(prefixStr + "_" + field.getProperty());
                            }
                        }
                    } else {
                        field.setColumn(property);
                    }
                }
                if (type != null) {
                    field.setType(type);
                } else {
                    field.setType("string");
                }
                fields.add(field);
            }
            if (fields.isEmpty()) {
                log.error("文件:{}中节点[rule]下没有字段映射,忽略掉该条规则", configFile);
                continue;
            }
            rule.setFields(fields);
            rulesHolder.putIfAbsent(String.format("%s_%s", namespace, id), rule);
        }
    }
}